#include "impl_pooled.h"

#include "messaging/drivers/drivers.h"


namespace evm { namespace messaging {

void PooledExecutor::start() {
    if (!stopped_)  // has started
        return;

    stopped_ = false;

    if (poller_.joinable())
        poller_.join();
    if (pool_.joinable()) {
        pool_.stop(true);  // redundant stop has no side effect
        pool_.join();
    }

    poller_ = thread([this]() {
        try {
            while (!stopped_) {
                auto incoming = listener->poll();
                if (!incoming)
                    continue;

                /* Here shared_ptr instances are copied and explicitly passed into lambda,
                 * to avoid GCC 4.7 nested lambda bug. */
                auto p = new pair<DispatcherPtr*, IncomingMsgPtr*>(
                                  new DispatcherPtr(dispatcher), new IncomingMsgPtr(incoming));

                bool was_submitted = pool_.post([](void *data) mutable {
                    auto p = static_cast<pair<DispatcherPtr*, IncomingMsgPtr*> *>(data);
                    auto dispatcher = p->first;
                    auto incoming = p->second;

                    (*dispatcher)->operator()(*incoming);

                    delete dispatcher;
                    delete incoming;
                    delete p;
                }, p);
                if (!was_submitted)  // pool size too small or incoming too frequent ?
                    log_error("Pooled executor, thread pool queue is full");
            }
        } catch (const exception& e) {
            log_info(e.what());
            throw;
        }
    });
    pool_.start();
}

void PooledExecutor::stop() {
    stopped_ = true;
}

void PooledExecutor::wait() {
    poller_.join();
    pool_.stop(true);
    pool_.join();
}

}}
